package com.sfx.mq.demo;

import com.sfx.mq.constants.MQConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.Date;
import java.util.List;

import static com.sfx.mq.constants.TopicProperties.RETRY_TIME_TOPIC;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-10-28
 * Time: 15:18
 */
public class RetryMessageTest {

    @Test
    public void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer-group");
        producer.setNamesrvAddr(MQConstant.NAME_SRV);
        producer.start();
        producer.setRetryTimesWhenSendFailed(2);
        producer.setRetryTimesWhenSendAsyncFailed(2);
        Message message = new Message(RETRY_TIME_TOPIC, "重试消息".getBytes());
        producer.send(message, 1000);
        System.out.println("发送消息成功");
        producer.shutdown();
    }

    /**
     * 消费者默认重试次数是16次
     * 顺序模式是int的最大值次,并发模式下是16次
     * 1. 如果重试了16次仍然失败怎么办 ?
     * 就认为这个消息是死信的消息,会放到死信队列中,队列名称就是 %DLQ% + 消费者组的名称
     * 1.1 第一种方案就是 再加一个消费者监听死信队列,如果有消息就保存到MySQL/文件中,给程序员发送短信,进行人工介入
     * 2. 能不能修改重试次数呢 ?
     * consumer.setMaxReconsumeTimes(3);
     *
     * @throws Exception
     */
    @Test
    public void consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-6");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        consumer.subscribe(RETRY_TIME_TOPIC, "*");
        // 设置最大重试次数
        consumer.setMaxReconsumeTimes(3);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                System.out.println("我是消费者,我消费失败,让生产者进行重试....");
                System.out.println("消息为 :" + new String(messageExt.getBody()));
                System.out.println("消费者重试次数为 : " + messageExt.getReconsumeTimes());
                System.out.println(new Date());
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
        System.in.read();
    }

    /**
     * 处理死信队列中的消息 :
     * 第一种方案就是 再加一个消费者监听死信队列,如果有消息就保存到MySQL/文件中,给程序员发送短信,进行人工介入
     * 缺点就是,如果有多个Topic,那么我们就需要写多个消费者 + 监听器-->写多个死信监听器
     *
     * @throws Exception
     */
    @Test
    public void DLQConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-6");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        consumer.subscribe("%DLQ%consumer-group-6", "*");
        // 设置最大重试次数
        consumer.setMaxReconsumeTimes(3);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                System.out.println("记录下来该消息,比如mysql,文件.....");
                System.out.println("人工处理该消息,给程序猿发送短信");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }

    /**
     * 第二种方案就是我们可以进行判断重试次数,达到一定的重试次数,进行记录消息,人工介入
     *
     * @throws Exception
     */
    @Test
    public void consumer2() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group-6");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        consumer.subscribe(RETRY_TIME_TOPIC, "*");
        // 设置最大重试次数
        consumer.setMaxReconsumeTimes(3);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = null;
                try {
                    // 这里进行业务处理,抛出异常进行重试
                    // 业务报错,返回null,返回RECONSUME_LATER 都会进行重试
                    messageExt = list.get(0);
                    handle(messageExt);
                } catch (Exception e) {
                    int reconsumeTimes = messageExt.getReconsumeTimes();
                    if (reconsumeTimes >= 3) {
                        // 达到一定的重试次数
                        System.out.println("记录下来该消息,比如mysql,文件.....");
                        System.out.println("人工处理该消息,给程序猿发送短信");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    // 否则进行重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            private void handle(MessageExt messageExt) {
                int x = 9 / 0;
            }
        });
        consumer.start();
        System.in.read();
    }
}